package net.bwie.realtime.dy.dwd.log.funcation.ql;



import net.bwie.realtime.dy.dwd.log.bean.TikTokBean;
import net.bwie.realtime.dy.dwd.log.bean.UserCountBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.util.ArrayList;

/**
 * @ClassName：t1
 * @Author: Vvn
 * @Date: 2025/5/27 11:25
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

public class realTimeUserCountFunction implements
        WindowFunction<TikTokBean, UserCountBean, String, TimeWindow> {

    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s,
                      TimeWindow window,
                      Iterable<TikTokBean> input,
                      Collector<UserCountBean> out) throws Exception {

        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());

        int UserCount = 0;

        // 获取到每个直播间下的用户日志数据
        for (TikTokBean bean : input) {

            String eventType = bean.getEventType();
            if ("enter".equals(eventType)) {
                UserCount++;
            }
            if ("leave".equals(eventType)) {
                UserCount--;
            }
        }


        out.collect(new UserCountBean(
                windowStartTime,
                windowEndTime,
                s,
                UserCount,
                System.currentTimeMillis()
        ));
    }
}
